package com.small.nacos.naming.consistency.ephemeral.distro;

import com.alibaba.nacos.api.exception.NacosException;
import com.alibaba.nacos.common.utils.Pair;
import com.alibaba.nacos.common.utils.StringUtils;
import com.small.nacos.naming.consistency.*;
import com.small.nacos.naming.core.Instances;
import com.small.nacos.naming.misc.GlobalExecutor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;

/**
 * @Author zhoujin
 * @Date 2022/10/29 12:48
 */
@Service("distroConsistencyService")
public class DistroConsistencyServiceImpl implements ConsistencyService{

    private final static Logger LOGGER = LoggerFactory.getLogger(DistroConsistencyServiceImpl.class);

    private final Map<String, ConcurrentLinkedQueue<RecordListener>> listeners = new ConcurrentHashMap<>();

    private final Notifier notifier = new Notifier();

    private final DataStore dataStore = new DataStore();

    @PostConstruct
    public void init() {
        GlobalExecutor.submitDistroNotifyTask(notifier);
    }

    @Override
    public void put(String key, Object value) throws NacosException {
        onPut(key,value);
    }

    @Override
    public void remove(String key) throws NacosException {
        onRemove(key);
        listeners.remove(key);
    }

    @Override
    public Datum get(String key) throws NacosException {
        return dataStore.get(key);
    }

    @Override
    public void listen(String key, RecordListener listener) throws NacosException {
        ConcurrentLinkedQueue<RecordListener> recordListeners = listeners.get(key);
        if (recordListeners == null){
            recordListeners = new ConcurrentLinkedQueue();
            ConcurrentLinkedQueue<RecordListener> recordListenersExist;
            if ((recordListenersExist = listeners.putIfAbsent(key, recordListeners)) != null) {
                recordListeners = recordListenersExist;
            }
        }
        if (!recordListeners.contains(listener)){
            synchronized (recordListeners){
                if (!recordListeners.contains(listener)){
                    recordListeners.add(listener);
                }
            }
        }

    }

    @Override
    public void unListen(String key, RecordListener listener) throws NacosException {
        ConcurrentLinkedQueue<RecordListener> recordListeners = listeners.get(key);
        if (recordListeners == null){
            return;
        }
        synchronized (recordListeners) {
            for (RecordListener recordListener : recordListeners) {
                if (recordListener.equals(listener)) {
                    recordListeners.remove(listener);
                    break;
                }
            }
        }
    }

    private void onPut(String key, Object value) {
        if (KeyBuilder.matchEphemeralKey(key)) {
            Datum datum = new Datum();
            datum.key = key;
            datum.value = value;
            dataStore.put(key,datum);
        }
        //监听器如果不包含当前实例就不添加任务
        if (!listeners.containsKey(key)) {
            return;
        }
        notifier.addTask(key,DataOperation.CHANGE);

    }

    public void onRemove(String key) {

        dataStore.remove(key);

        if (!listeners.containsKey(key)) {
            return;
        }

        notifier.addTask(key, DataOperation.DELETE);
    }


    class Notifier implements Runnable{

        ConcurrentHashMap<String,String> services = new ConcurrentHashMap();

        private BlockingQueue<Pair<String, DataOperation>> tasks = new ArrayBlockingQueue<>(1024 * 1024);

        public void addTask(String datumKey, DataOperation action){
            //已包含当前任务
            if (services.contains(datumKey) && action == DataOperation.CHANGE){
                return;
            }
            if (action == DataOperation.CHANGE){
                services.put(datumKey, StringUtils.EMPTY);
            }
            tasks.offer(Pair.with(datumKey, action));
        }

        @Override
        public void run() {
            LOGGER.info("distro notifier started");
            for (; ; ) {
                try {
                    Pair<String, DataOperation> pair = tasks.take();
                    handle(pair);
                } catch (Throwable e) {
                    LOGGER.error("[NACOS-DISTRO] Error while handling notifying task", e);
                }
            }
        }

        private void handle(Pair<String, DataOperation> pair) {

            String datumKey = pair.getFirst();
            DataOperation action = pair.getSecond();

            services.remove(datumKey);
            ConcurrentLinkedQueue<RecordListener> recordListeners = listeners.get(datumKey);
            if (recordListeners == null) {
                LOGGER.info("[DISTRO-WARN] RecordListener not found, key: {}", datumKey);
                return;
            }
            for (RecordListener listener : recordListeners) {
                try {
                    if (action == DataOperation.CHANGE){
                        Datum<Instances> datum = dataStore.get(datumKey);
                        if (datum != null){
                            listener.onChange(datumKey,datum.value);
                        } else {
                            LOGGER.info("[DISTRO-WARN] data not found, key: {}", datumKey);
                        }

                    }else if (action == DataOperation.DELETE){
                        listener.onDelete(datumKey);
                    }
                }catch (Throwable e) {
                    LOGGER.error("[NACOS-DISTRO] error while notifying listener of key: {}", datumKey, e);
                }
            }
        }
    }
}
